package com.edu360.select

import java.sql.DriverManager

import com.edu360.utils.ToMysqlUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SelectProCityCountRDD {
  def main(args: Array[String]): Unit = {
    if (args.length != 1) {
      println(
        """
          |cn.dmp.tools.Bzip2Parquet
          |参数：
          | logInputPath
        """.stripMargin)
      sys.exit()
    }
    // 1 接受程序参数
    val Array(logInputPath) = args

    // 2 创建sparkconf->sparkContext
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    // RDD 序列化到磁盘 worker与worker之间的数据传输
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)

    // 3 读取日志数据
    val rawdata = sc.textFile(logInputPath)

    val filter: RDD[Array[String]] = rawdata
      .map(line => line.split(",", line.length))
      .filter(_.length >= 85)
    val proAndCityAndOne = filter.map(arr => {
      val province = arr(24)
      val city = arr(25)
      ((province, city), 1)
    })
    val reduced: RDD[((String, String), Int)] = proAndCityAndOne.reduceByKey(_+_)

   reduced.foreachPartition(iter=>ToMysqlUtils.data2MySQL(iter))

    sc.stop()
  }
}
